package net.heartbeat.handler;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;

import net.heartbeat.HeartbeatTransferer;

/**
 * 任务类：接收心跳消息
 * 
 * @author Bai Yu
 */
class ReceiveHeartbeat<HeartbeatType> implements Runnable {
	/**
	 * 接收心跳消息
	 * 
	 * @param transferer
	 *            心跳包的接收策略
	 * @param receiver
	 *            接收心跳消息的底层实现
	 * @param abscenceDetecter
	 *            收集参数：统计收到了来自哪些对象的心跳包
	 * @param parsedHeartbeats
	 *            收集参数：收集收到的心跳消息，供后续处理<br>
	 *            可以为null，代表不需要收集心跳消息
	 */
	public ReceiveHeartbeat(HeartbeatTransferer<HeartbeatType> transferer, HeartbeatReceiver receiver,
			HeartbeatAbsenceDetecter abscenceDetecter,
			BlockingQueue<ParsedHeartbeatMessage<HeartbeatType>> parsedHeartbeats) {
		this.transferer = transferer;
		this.receiver = receiver;
		this.abscenceDetecter = abscenceDetecter;
		this.parsedHeartbeats = parsedHeartbeats;
	}

	@Override
	public void run() {
		while (true) {
			HeartbeatMessage message;
			try {
				message = receiver.accept();
			} catch (FailedSourceException e) {
				continue;
			} catch (IOException e) {
				break;
			}

			try {
				HeartbeatType heartbeat = transferer.receive(message.inputStream);
				synchronized (abscenceDetecter) {
					abscenceDetecter.onHeartbeat(message.source);
				}
				if (parsedHeartbeats != null) {
					parsedHeartbeats.add(new ParsedHeartbeatMessage<>(message.source, heartbeat));
				}
			} catch (IOException e) {
			} finally {
				try {
					message.inputStream.close();
				} catch (IOException e) {
				}
			}
		}
	}

	private HeartbeatTransferer<HeartbeatType> transferer;
	private HeartbeatReceiver receiver;

	private HeartbeatAbsenceDetecter abscenceDetecter;
	private BlockingQueue<ParsedHeartbeatMessage<HeartbeatType>> parsedHeartbeats;

}
